package com.ylkj.bus.dispatcher.handler;


import com.hwl.redis.RedisConfig;
import com.ylkj.bus.MessageContent;
import com.ylkj.bus.device.onoffline.model.OnOfflineDeviceEnty;
import com.ylkj.bus.device.report.DeviceOnlineStatusProcess;
import com.ylkj.bus.device.report.UpdateDevicReportAndUnpack;
import com.ylkj.bus.dispatcher.IStreamHandler;
import com.ylkj.dao.RdsConfig;
import com.ylkj.dao.RedisConfigInfo;
import lombok.Data;
import lombok.RequiredArgsConstructor;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.util.OutputTag;



/**
 * @Author: wengzx
 * @Date: 2025/4/21
 * @Desc:
 */
@RequiredArgsConstructor
@Data
public class DeviceOnlineDataStreamHandler implements IStreamHandler {
    private final String rdsTableName;
    private final RdsConfig rdsConfig;
    private final RedisConfig redisConfig;
    private final String rdsTableName2;
    private final String rdsTableName3;
    private final String rdsTableName4;
    private final String rdsTableName5;
    private final String rdsTableName6;
    private final RedisConfigInfo redisConfig1;
    @Override
    public void handle(DataStream<MessageContent> stream) {
        // 定义侧输出流标签
        OutputTag<OnOfflineDeviceEnty> outTagStatus = new OutputTag<OnOfflineDeviceEnty>("outTagStatus") {};
        //定义主流
        SingleOutputStreamOperator<MessageContent> deviceOnlineStatusProcess = stream.process(new DeviceOnlineStatusProcess(outTagStatus,rdsConfig,rdsTableName5,rdsTableName6,redisConfig1)).name("device_online_status_unpack_process");
        //获取侧输出流
        DataStream<OnOfflineDeviceEnty> sideOutputDs = deviceOnlineStatusProcess.getSideOutput(outTagStatus);
        sideOutputDs.print();
        //更新设备档案表
        sideOutputDs.process(new UpdateDevicReportAndUnpack(rdsTableName, rdsConfig,redisConfig,rdsTableName2,rdsTableName3)).name(rdsTableName + "_update_process");

        //sideOutputDs.print();
    }
}
